package com.liuguang.framework.base;


import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;
import io.reactivex.rxjava3.subjects.PublishSubject;

public class RxBus {
    private static volatile RxBus sInstance;
    private final PublishSubject<Object> mEventBus = PublishSubject.create();

    class Message {
        int code;
        Object event;

        public Message(int i, Object obj) {
            this.code = i;
            this.event = obj;
        }
    }

    public static RxBus getInstance() {
        if (sInstance == null) {
            synchronized (RxBus.class) {
                if (sInstance == null) {
                    sInstance = new RxBus();
                }
            }
        }
        return sInstance;
    }

    public void post(Object obj) {
        this.mEventBus.onNext(obj);
    }

    public void post(int i, Object obj) {
        this.mEventBus.onNext(new Message(i, obj));
    }

    public Observable toObservable() {
        return this.mEventBus;
    }

    public <T> Observable<T> toObservable(Class<T> cls) {
        return this.mEventBus.ofType(cls);
    }

    public <T> Observable<T> toObservable(final int i, final Class<T> cls) {
        return this.mEventBus.ofType(Message.class).filter(new Predicate<Message>() {
            public boolean test(Message message) throws Exception {
                return message.code == i && cls.isInstance(message.event);
            }
        }).map(new Function<Message, T>() {
            public T apply(Message message) throws Exception {
                return (T) message.event;
            }
        });
    }
}
